package com.xiehua.task.actor;

import akka.actor.AbstractActor;
import akka.actor.ActorSelection;
import akka.pattern.Patterns;
import com.xiehua.task.actor.cmd.Cmd;
import com.xiehua.task.job.bucket.ReadyBucket;
import com.xiehua.task.job.pool.Job;
import com.xiehua.task.redis.RedisTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;

import java.util.List;

public class TaskDisposeActor extends AbstractActor {


    private static final Logger LOGGER = LoggerFactory.getLogger(TaskNotifyActor.class);

    @Override
    public Receive createReceive() {
        return receiveBuilder().match(Cmd.class, t -> t.getCmd().equals(Cmd.EVT_TASK_DISPOSE), s -> {
            if (LOGGER.isDebugEnabled()) LOGGER.debug("收到消息type：{}", s.getCmd());
            ReadyBucket readyBucket = new ReadyBucket(RedisTemplate.getJedis());
            List<Job> readyJob = readyBucket.pollTaskBatch();
            readyBucket.clear();
            LOGGER.info("通知客户端处理消息条数:{}", readyJob.size());
            // 将消息发给下一个处理者TaskNotifyActor
            ActorSelection taskNotifyActor = getContext().actorSelection("/user/TaskNotifyActor");
            //taskNotifyActor.tell(cmd,self());
            dispatcher(readyJob, taskNotifyActor);
        }).match(Cmd.class, t -> t.getCmd().equals(Cmd.CMD_TASK_NOTIFY_ASYNC_RESP), s -> {
            LOGGER.info("JOB已通知客户端:{}", s.getT());
            //deleteSnapshot();
        }).matchAny(t -> {
            LOGGER.error("没有对应Actor处理本次消息~");
            unhandled(t);
        }).build();
    }

    /**
     * 任务分发~
     **/
    private void dispatcher(List<Job> readyJob, ActorSelection taskNotifyActor) {
        readyJob.forEach(s -> {
            Cmd cmd = new Cmd(Cmd.CMD_TASK_NOTIFY, s);
            Future<Object> future = Patterns.ask(taskNotifyActor, cmd, 2000);
            Patterns.pipe(future, getContext().getSystem().dispatcher()).to(getSelf());
        });
    }
}
